package com.sink;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author 谢秉均
 * @date 2024/10/11--1:33
 */
public class sinkMysql {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<String> result = env.socketTextStream("192.168.44.201",7777);

        SinkFunction<String> sink = JdbcSink.sink(
                "insert  into ws (id,ts) values (?,?)",
                new JdbcStatementBuilder<String>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                        preparedStatement.setString(1, "12");
                        preparedStatement.setString(2, s);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withMaxRetries(3)
                        .withBatchSize(100)
                        .withBatchIntervalMs(3000)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withConnectionCheckTimeoutSeconds(60) //单位s
                        .withUsername("root")
                        .withPassword("1234")
                        .withUrl("jdbc:mysql://localhost:3306/message?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8")
                        .build()
        );

        result.addSink(sink);
        env.execute();
    }
}
